package ru.nsu.fit.iconveyor.engine.smart;

import ru.nsu.fit.iconveyor.conveyor.exception.NodeExecutionException;
import ru.nsu.fit.iconveyor.conveyor.node.Node;

import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author Timur Zolotuhin (tzolotuhin@gmail.com)
 */
public class Executor implements  Runnable {
    private static final int TASK_QUEUE_CAPACITY = 1000;

    private Collection<Observer> observers = new ArrayList<Observer>();
    private BlockingQueue<Task> tasksQueue = new ArrayBlockingQueue<Task>(TASK_QUEUE_CAPACITY);

    // Main data
    private boolean available = true;
    private final int number;

    private final Object engineMutex;

    // Mutex
    private final Object generalMutex = new Object();

    public Executor(int number, Object engineMutex) {
        this.number = number;
        this.engineMutex = engineMutex;
    }

    public boolean isAvailable() {
        synchronized (generalMutex) {
            return available;
        }
    }

    public int getNumber() {
        return number;
    }

    /**
     * Tries to add new task
     * @param task
     * @throws IllegalStateException if queue is overflown
     * @return
     */
    public boolean addTask(Task task) {
        synchronized (generalMutex) {
            if (!available)
                return false;

            tasksQueue.add(task);

            available = false;
            return true;
        }
    }

    @Override
    public void run() {
        boolean stop = false;
        while(true) {
            synchronized (generalMutex) {
                if (stop)
                    break;
            }

            Task currentTask;
            try {
                currentTask = tasksQueue.poll(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
            } catch (InterruptedException e) {
                stop = true;
                continue;
            }

            boolean  localAvailable = true;
            synchronized (generalMutex) {
                localAvailable = available;
            }

            if (!localAvailable) {
                try {
                    Map<String, Object> result = currentTask.getNode().execute(currentTask.getInput(), currentTask.getProp());
                    currentTask.getCell4Result().complete(result);
                } catch (NodeExecutionException e) {
                    notifyObserversOnError(e);
                }
                synchronized (generalMutex) {
                    available = true;
                }
                synchronized (engineMutex) {
                    engineMutex.notify();
                }
            }
        }
    }

    public static interface Observer {
        void onError(Exception cause);
    }

    private void notifyObserversOnError(NodeExecutionException cause) {
        for(Observer observer : observers) {
            observer.onError(cause);
        }
    }

    public void addObserver(Observer observer) {
        observers.add(observer);
    }

    public void removeObserver(Observer observer) {
        observers.remove(observers);
    }

    public static class Task {
        Node node;
        Map<String, Object> input;
        Cell4Result cell4Result;
        Properties prop;

        public Node getNode() {
            return node;
        }

        public void setNode(Node node) {
            this.node = node;
        }

        public Map<String, Object> getInput() {
            return input;
        }

        public void setInput(Map<String, Object> input) {
            this.input = input;
        }

        public Cell4Result getCell4Result() {
            return cell4Result;
        }

        public void setCell4Result(Cell4Result cell4Result) {
            this.cell4Result = cell4Result;
        }

        public Properties getProp() {
            return prop;
        }

        public void setProp(Properties prop) {
            this.prop = prop;
        }
    }
}
